import com.opdar.athena.client.agent.base.Configuration;
import com.opdar.athena.entities.MessageEntity;
import com.opdar.athena.entities.UserMigrateEntity;
import com.opdar.athena.messages.HeartbeatMessage;
import com.opdar.athena.utils.JSONUtil;
import com.opdar.athena.base.MessageType;
import com.opdar.athena.messages.UserMessage;

import java.io.IOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by shiju on 2017/6/29.
 */
public class ClientSocket {

    private ExecutorService es = Executors.newFixedThreadPool(3);
    final byte[] project = Configuration.getConfiguration().getProject().getBytes();

    ReadRunnable readRunnable;
    Socket socket;
    String token;
    public ClientSocket(String address,int port) {
        try {
            socket = new Socket(address, port);
            socket.setSoTimeout(60000);
            socket.setSoLinger(true,5000);
            es.execute(readRunnable = new ReadRunnable(socket,project, new MessageCallback() {
                @Override
                public void migrate(UserMigrateEntity migrateMessage) {
                    System.out.println("Migrate id :"+migrateMessage.getId());
                    System.out.println("Migrate destination :"+migrateMessage.getDestination());
                    System.out.print("正在准备迁移...");
                    String[] destination = migrateMessage.getDestination().split(":");
                    destory();
                    new ClientSocket(destination[0],Integer.valueOf(destination[1])).bindWithMigrate(migrateMessage.getId(),token);
                }

                @Override
                public void heartbeat(HeartbeatMessage heartbeatMessage) {
                    ClientSocket.this.heartbeat();
                }

                AtomicLong atomicLong = new AtomicLong(0);
                @Override
                public void callback(MessageEntity message) {
                    System.out.println(atomicLong.getAndIncrement());
                }

                @Override
                public void onClose() {

                }
            }));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public ClientSocket heartbeat(){
        es.execute(new HeartRunnable(socket,project));
        return this;
    }
    public ClientSocket heartbeatSync(){
        ByteBuffer buffer = ByteBuffer.allocate(project.length + 18);
        buffer.put(project);
        buffer.putInt(1);
        buffer.putShort((short) MessageType.HEARTBEAT);
        buffer.putInt(8);
        buffer.putLong(System.currentTimeMillis());
        try {
            socket.getOutputStream().write(buffer.array());
            socket.getOutputStream().flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return this;
    }
    public ClientSocket bindWithMigrate(String migrateId,String token){
        try {
            this.token = token;
            UserMessage userMessage = new UserMessage();
            userMessage.setCommand("BindWithMigrate");
            userMessage.setToken(token);
            userMessage.setMigrateId(migrateId);
            byte[] user = JSONUtil.toJSONAsBytes(userMessage);
            ByteBuffer buffer = ByteBuffer.allocate(project.length + 10+user.length);
            buffer.put(project);
            buffer.putInt(1);
            buffer.putShort((short) MessageType.USER);
            buffer.putInt(user.length);
            buffer.put(user);
            socket.getOutputStream().write(buffer.array());
            System.out.println("迁移成功.");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("迁移失败.");
        }
        return this;
    }

    public ClientSocket bindWithToken(String token){
        this.token = token;
        UserMessage userMessage = new UserMessage();
        userMessage.setCommand("BindWithToken");
        userMessage.setToken(token);
        byte[] user = JSONUtil.toJSONAsBytes(userMessage);
        ByteBuffer buffer = ByteBuffer.allocate(project.length + 10+user.length);
        buffer.put(project);
        buffer.putInt(1);
        buffer.putShort((short) MessageType.USER);
        buffer.putInt(user.length);
        buffer.put(user);
        try {
            socket.getOutputStream().write(buffer.array());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return this;
    }

    private void destory() {
        try {
            readRunnable.stop();
            socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        es.shutdown();
    }

    public static void main(String[] args) {
        long time = System.currentTimeMillis();
        ClientSocket clientSocket = new ClientSocket("192.168.1.178",18181).bindWithToken("0260f151-e994-44ab-a51c-f131640ac79e");
    }

}
